package com.flying.learning.si.sikg.client;

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.config.EnableIntegration;
import org.springframework.integration.dsl.IntegrationFlow;
import org.springframework.integration.dsl.IntegrationFlows;
import org.springframework.integration.kafka.dsl.Kafka;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.ProducerFactory;
import org.springframework.kafka.listener.ConcurrentMessageListenerContainer;
import org.springframework.kafka.listener.ContainerProperties;
import org.springframework.kafka.requestreply.ReplyingKafkaTemplate;

@Configuration
public class GatewayConfig {
    // Client side
//    @MessagingGateway//(defaultReplyChannel = "kafkaReplies", defaultReplyTimeout = "10000")
//    @Lazy
    public interface Exchanger {
        //@Gateway@Payload
        String exchange(String out);
    }

    // Listener Container to be set up in ReplyingKafkaTemplate
    @Bean
    public ReplyingKafkaTemplate replyingKafkaTemplate(ProducerFactory producerFactory, ConsumerFactory consumerFactory) {
        ContainerProperties containerProperties = new ContainerProperties("TRANSACTION_RESULT3");
        ConcurrentMessageListenerContainer<String, String> container = new ConcurrentMessageListenerContainer<String, String>(consumerFactory,
                containerProperties);
        return new ReplyingKafkaTemplate(producerFactory, container);
    }

    @Configuration
    @EnableIntegration
    public static class SIConfig {
        @Bean
        public IntegrationFlow outboundGateFlow(ReplyingKafkaTemplate<String, String, String> kafkaTemplate) {
//        return IntegrationFlows.from("kafkaRequests")
//                .handle(Kafka.outboundGateway(kafkaTemplate).topic("TRANSACTION_TOPIC3"))
//                .channel("kafkaReplies")
//                .get();
            return IntegrationFlows.from(Exchanger.class)
                    .handle(Kafka.outboundGateway(kafkaTemplate).topic("TRANSACTION_TOPIC3"))
                    .get();
        }
    }
}
